package com.xiewei.kafka.chapter.chapter;

import kafka.utils.ShutdownableThread;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.*;

@Slf4j
public class Consumer  extends ShutdownableThread {
    private KafkaConsumer comsumer;

    private List<ConsumerRecord> list = new ArrayList<>();

    public Consumer() {
        super("ConsumerTest", false);
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        //groupId 所属的分组
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test2");
        //是否自动提交offset
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        //自动提交时间间隔
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        //设置使用最早的offset偏移量为当前 消费
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        //设置心跳时间
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        //key value 序列化 反序列化
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        this.comsumer =new KafkaConsumer(props);
    }

    @Override
    public void doWork() {
        //设置topic
        //omsumer.assign(Arrays.asList(new TopicPartition("MyTopic",0)));  指定消费分区 和 subscribe互斥
        comsumer.subscribe(Collections.singletonList("MyTopic"));
        ConsumerRecords<Integer,String> records = comsumer.poll(1000);
        for (ConsumerRecord record : records){
            System.out.println("partition:"+record.partition()+"receiver messages:["+record.key()+"->" +record.value()+"],offset:"+record.offset());

            list.add(record);

        }
//        if (list.size() >= 5) {
//            log.info("commit offset");
//            comsumer.commitSync();
//            list.clear();
//        }

    }

    public static void main(String[] args) {
        Consumer consumer =new Consumer();
        consumer.start();

    }
}
